package com.tonypacer.hadooplearn.mapreduce;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.text.DecimalFormat;

/**
 * 程序名: 08.求平均数
 *  
 
 时间	新增	总确诊	治愈	死亡
2020.1.20	72	270	25	6
2020.1.21	105	375	28	9
2020.1.22	69	444	28	17
2020.1.23	105	549	31	24
2020.1.24	180	729	32	39
2020.1.25	323	1052	42	52
2020.1.26	371	1423	44	76
2020.1.27	1291	2714	47	100
2020.1.28	840	3554	80	125
2020.1.29	1032	4586	90	162
2020.1.30	1220	5806	116	204
2020.1.31	1347	7153	166	249
2020.2.01	1921	9074	215	294
2020.2.02	2103	11177	295	350
2020.2.03	2345	13522	396	414
2020.2.04	3156	16678	520	479
2020.2.05	2987	19655	633	549
2020.2.06	2447	22112	817	618

 */
public class Average {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        job.setJobName("Average");
        job.setJarByClass(Average.class);

        job.setMapperClass(AverageMapper.class);
        job.setReducerClass(AverageReducer.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(DoubleWritable.class);

        FileInputFormat.addInputPath(job,new Path("InputResource/average"));

        FileSystem fs = FileSystem.get(conf);
        Path outdir = new Path("Output/Average_output");
        if (fs.exists(outdir)){
            fs.delete(outdir,true);
        }
        FileOutputFormat.setOutputPath(job,outdir);
        boolean success = job.waitForCompletion(true);
        if (success) {
            System.out.print("success!");
        }
    }
}

class AverageMapper extends Mapper<LongWritable,Text,Text,DoubleWritable>{
    Text keyInfo = new Text();
    DoubleWritable valueInfo = new DoubleWritable();
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String[] strs = value.toString().split(" ");
        keyInfo.set(strs[0]);
        valueInfo.set(Double.parseDouble(strs[1]));
        context.write(keyInfo,valueInfo);
    }
}

class AverageReducer extends Reducer<Text,DoubleWritable,Text,Text>{
    Text valueInfo = new Text();
    double average;
    @Override
    protected void reduce(Text key, Iterable<DoubleWritable> values, Context context) throws IOException, InterruptedException {
        double sum = 0;
        int count = 0;
        for (DoubleWritable value: values){
            sum += value.get();
            count ++;

        }
        average = sum / count;
        DecimalFormat df = new DecimalFormat("#.00");
        valueInfo.set(df.format(average));
        context.write(key,valueInfo);
    }
}



